package com.SparkCore.RDD.persist

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 * 持久化
 *
 */
object Spark04_RDD_Persist {
  def main(args: Array[String]): Unit = {
    /**
     * cache 将数据临时存储在欸村中及逆行数据重用
     *       会在血缘关系中添加新的依赖，一旦出现了问题可以从头读取数据
     * persist 将数据临时存储在磁盘文件中进行数据重用
     *         涉及到磁盘IO，性能较低，但是数据安全
     *         如果作业执行完毕，临时保存的数据文件就会丢失
     *
     * checkpoint：将数据长久的保存在磁盘文件中进行数据重用
     *             涉及到磁盘IO，性能较低，到那时数据安全
     *             为保证安全，一般情况下哎会独立执行作业
     *             为了能够提高效率，一般情况下，时需要和cache联合使用
     *             在执行过程中会切点血缘关系，重新建立新的血缘关系
     *             checkpoint会切断血缘关系，建立新的血缘关系
     *             等同于我们的数据源
     */
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("WordCount")
    val sc: SparkContext = new SparkContext(sparkConf)
    sc.setCheckpointDir("cp")

    //创建RDD算子
    val path = "datas/1.txt"
    val txt: RDD[String] = sc.textFile(path)

    val flatRDD: RDD[String] = txt.flatMap(_.split(" "))

    val mapRDD: RDD[(String, Int)] = flatRDD.map(word => {
      println("+++++++++++")
      (word, 1)
    })

    mapRDD.checkpoint()

    val reduceRDD: RDD[(String, Int)] = mapRDD.reduceByKey(_+_)

    reduceRDD.collect().foreach(println)

    println("====================")

    val reduceRDD1: RDD[(String, Iterable[Int])] = reduceRDD.groupByKey()

    reduceRDD1.collect().foreach(println)
    //关闭链接
    sc.stop()
  }
}
